package com.jvxb.manage.livable.service.dbQuery.impl;

import com.baomidou.mybatisplus.core.conditions.Condition;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.jvxb.manage.livable.entity.AssetsInstance;
import com.jvxb.manage.livable.service.AssetsInstanceService;
import com.jvxb.manage.livable.service.dbQuery.RedisQueryService;
import com.jvxb.manage.livable.util.RedisCmdAnalysisUtil;
import com.jvxb.manage.livable.util.RedisUtil;
import com.jvxb.modules.configuration.security.SecurityUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import redis.clients.jedis.Protocol;
import redis.clients.jedis.exceptions.JedisConnectionException;
import redis.clients.util.RedisInputStream;
import redis.clients.util.RedisOutputStream;
import redis.clients.util.SafeEncoder;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;

/**
 * @author jvxb
 * @since 2022-02-08
 */
@Service
public class RedisQueryServiceImpl implements RedisQueryService {

    @Autowired
    AssetsInstanceService assetsInstanceService;

    @Override
    public String query(Long clusterId, String command) throws Exception {
        //1）检查是否支持当前命令
        if (!supportCmd(command)) {
            throw new RuntimeException("不支持当前命令！");
        }
        //2）获取实例
        AssetsInstance assetsInstance = assetsInstanceService.getOne(Condition.<AssetsInstance>create().eq(AssetsInstance.CLUSTER_ID, clusterId).last("limit 1"));
        if (assetsInstance == null) {
            throw new RuntimeException("没有可用实例！集群id=" + clusterId);
        }
        //3）查询结果
        String auth = null;
        String result = execute(assetsInstance.getHost(), Integer.parseInt(assetsInstance.getPort()), command, auth);
        if ("".equals(result) || null == result) {
            result = "<nil>";
        }
        if (result.startsWith("MOVED")) {
            String[] s1 = result.split(" ");
            String movedAddr = s1[s1.length - 1];
            String[] movedAddrSplit = movedAddr.split(":");
            result = execute(movedAddrSplit[0], Integer.parseInt(movedAddrSplit[1]), command, auth);
        }
        //4）保存查询记录
        saveQueryRecord(clusterId, command);
        return result;
    }

    @Override
    public void getRedisQueryRecord(Long clusterId, IPage page) {
        Long userId = SecurityUtil.currentUserId();
        String queryRecordKey = "query:redis:" + userId + ":" + clusterId;
        int begin = (int) ((page.getCurrent() - 1) * page.getSize());
        int end = (int) (begin + page.getSize() - 1);
        Set set = RedisUtil.zReverseRange(queryRecordKey, begin, end);
        page.setRecords(new ArrayList(set));
        page.setTotal(RedisUtil.zcard(queryRecordKey));
    }

    private void saveQueryRecord(Long clusterId, String command) {
        Long userId = SecurityUtil.currentUserId();
        String queryRecordKey = "query:redis:" + userId + ":" + clusterId;
        RedisUtil.zSet(queryRecordKey, command, System.currentTimeMillis());
        //保存50条记录，保存30天
        if (RedisUtil.zcard(queryRecordKey) > 50) {
            RedisUtil.zRemoveRange(queryRecordKey, 50, -1);
        }
        RedisUtil.expire(queryRecordKey, 30 * 24 * 60 * 60);
    }

    private boolean supportCmd(String command) {
        //todo 检测其它命令。
        if (command.trim().startsWith("keys")) {
            return false;
        }
        return true;
    }

    private String execute(String ip, Integer port, String command, String auth) throws Exception {
        //close方法关闭Socket连接后，Socket对象所绑定的端口并不一定马上释放
        //系统有时在Socket连接关闭才会再确认一下是否有因为延迟面未到达的数据包
        //避免重启时，有时候的port already bind
        Socket socket = new Socket();
        socket.setReuseAddress(true);
        socket.setKeepAlive(true);//保持TCP连接，不释放资源
        socket.setTcpNoDelay(true);//立即发送数据，不合并数据包
        socket.setSoLinger(true, 0);//强制关闭连接，不阻塞close(阻塞0s)
        socket.connect(new InetSocketAddress(ip, port), 3000);
        socket.setSoTimeout(3000);//读取数据阻塞超时时间3s(0是一直阻塞)
        OutputStream outputStream = socket.getOutputStream();
        InputStream inputStream = socket.getInputStream();
        RedisOutputStream ros = new RedisOutputStream(outputStream);
        RedisInputStream ris = new RedisInputStream(inputStream);
        //认证
        if (auth != null) {
            String authFormat = formart(auth);
            outputStream.write(authFormat.getBytes("UTF-8"));
            ris.readLine();
        }
        writeCmd(ros, command);

        try {
            Object read = Protocol.read(ris);
            return makeResult(read);
        } catch (Exception e) {
            if (e.getMessage().startsWith("MOVED")) {
                return e.getMessage();
            } else {
                throw new Exception(e);
            }
        } finally {
            socket.close();
        }
    }

    //解析redis协议
    public static String formart(String command) {
        String delSpaceString = command.replaceAll("\\s{1,}", " ");
        StringBuilder stringBuilder = new StringBuilder();
        String[] split = delSpaceString.split(" ");
        stringBuilder.append("*" + split.length + "\r\n");
        for (int i = 0; i < split.length; i++) {
            stringBuilder.append("$" + split[i].length() + "\r\n");
            stringBuilder.append(split[i] + "\r\n");
        }
        return stringBuilder.toString();
    }

    public static final byte DOLLAR_BYTE = '$';
    public static final byte ASTERISK_BYTE = '*';

    private String makeResult(Object read) {
        String tmp = "";
        if (read == null) {
            return tmp;
        } else if (read.getClass().equals(ArrayList.class)) {
            ArrayList list = (ArrayList) read;
            for (int i = 0; i < list.size(); i++) {
                Object object = list.get(i);
                tmp += makeResult(object);
                if (i != list.size() - 1) {
                    tmp += "\r\n";
                }
            }
        } else if (read.getClass().equals(Long.class)) {
            tmp = (Long) read + "";
        } else {
            tmp = new String((byte[]) read, Charset.forName("utf-8"));
        }
        return tmp;
    }

    private void writeCmd(RedisOutputStream ros, String ask) {
        List<String> tmp = RedisCmdAnalysisUtil.getInstance().analysisRedisCmd(ask);
        byte[] cmd = SafeEncoder.encode(tmp.get(0));
        try {
            ros.write(ASTERISK_BYTE);
            ros.writeIntCrLf(tmp.size());
            ros.write(DOLLAR_BYTE);
            ros.writeIntCrLf(cmd.length);
            ros.write(cmd);
            ros.writeCrLf();

            for (int i = 1; i < tmp.size(); i++) {
                byte[] arg = SafeEncoder.encode(tmp.get(i));
                ros.write(DOLLAR_BYTE);
                ros.writeIntCrLf(arg.length);
                ros.write(arg);
                ros.writeCrLf();
            }
            ros.flush();
        } catch (IOException e) {
            throw new JedisConnectionException(e);
        }
    }
}
